package com.smsc.headend.task.engine.service.calc.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.common.utils.UDateUtils;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.data.dto.ReCalcDailyEnergyDTO;
import com.smsc.headend.module.data.entity.RdEnergyDaily;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.request.entity.LogRequest;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.TaskReadData;
import com.smsc.headend.module.task.proto.TaskReadDataProto;
import com.smsc.headend.task.engine.mapper.LineLossMapper;
import com.smsc.headend.task.engine.mapper.RdEnergyDailyMapper;
import com.smsc.headend.task.engine.service.DataProcessService;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.calc.RdEnergyDailyService;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.driver.jdbc.core.datasource.ShardingSphereDataSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Instant;
import java.util.*;
import java.util.stream.Collector;
import java.util.stream.Collectors;

@Service
@Slf4j
public class RdEnergyDailyServiceImpl extends ServiceImpl<RdEnergyDailyMapper, RdEnergyDaily> implements RdEnergyDailyService {

    private final static Long ENERGY_DATA_ITEM_ID = 1020030004102L;

    @Autowired
    AssetService assetService;

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    DataSource dataSource;

    @Autowired
    DataProcessService dataProcessService;

    @Value("${spring.datasource.dbnames}")
    String[] dnNames;

    @Autowired
    LineLossMapper lineLossMapper;

    @Value("${spring.datasource.dbtype:oracle}")
    String dbType;

    @Override
    public void calcDailyEnergyInterval(TaskReadData taskReadData) {
        if (Objects.isNull(taskReadData) || CollectionUtils.isEmpty(taskReadData.getTaskMeterReads())) {
            return;
        }

        List<RdMeterReadsLp> energyDataList = taskReadData.getTaskMeterReads().stream().filter(rdMeterReadsLp -> {
            return ENERGY_DATA_ITEM_ID.equals(rdMeterReadsLp.getDataItemId())
                    && rdMeterReadsLp.getTv().equals(UDateUtils.roundToCurrentDay(rdMeterReadsLp.getTv()));
        }).distinct().collect(Collectors.toList());
        if (CollectionUtils.isEmpty(energyDataList)) {
            return;
        }
        log.debug("daily energy size {}", energyDataList.size());

        TaskReadData readData = new TaskReadData();
        readData.setTaskMeterReads(energyDataList);
        kafkaTemplate.send(TaskKafkaTopic.DAILY_ENERGY_DATA, Convert.toStr(energyDataList.get(0).getMeterId()),
                ProtoBufferUtil.toProtoBuffer(readData, TaskReadDataProto.TaskReadData.class).toByteArray());
    }

    @Override
    public void calcDailyEnergyInterval(List<RdMeterReadsLp> dailyEnergyList, boolean needLastDay) {
        List<RdEnergyDaily> rdEnergyDailyList = new ArrayList<>();

        Map<Long, List<RdMeterReadsLp>> meterDailyEnergyMap = dailyEnergyList.stream().collect(
                Collectors.groupingBy(RdMeterReadsLp::getMeterId, toSortedList(Comparator.comparing(RdMeterReadsLp::getTv)))
        );
        meterDailyEnergyMap.forEach((meterId, energyList) -> {
            if (CollectionUtils.isEmpty(energyList)) {
                return;
            }
            if (needLastDay) {
                RdMeterReadsLp rdMeterReadsLp = energyList.get(0);
                long lastDay = rdMeterReadsLp.getTv() - 86400;
                List<RdMeterReadsLp> lastDayEnergyList = dataProcessService.getRdMeterReadLpList(meterId, rdMeterReadsLp.getLpDataItemId(),
                        rdMeterReadsLp.getDataItemId(), lastDay, lastDay);
                if (CollectionUtils.isNotEmpty(lastDayEnergyList)) {
                    lastDayEnergyList = lastDayEnergyList.stream()
                            .filter(d -> d.getTv().equals(UDateUtils.roundToCurrentDay(d.getTv())))
                            .collect(Collectors.toList());
                    energyList.addAll(lastDayEnergyList);
                }

                energyList = energyList.stream().collect(toSortedList(Comparator.comparing(RdMeterReadsLp::getTv)));
            }
            Meter meter = assetService.getMeterById(meterId);
            if (Objects.isNull(meter)) {
                log.error("Meter not exists. meterId={}", meterId);
                return;
            }
            double ct = parseCtOrVtValue(meter.getCt());
            double vt = parseCtOrVtValue(meter.getVt());
            double scale = ct * vt;

            for (int i = 1; i < energyList.size(); i++) {
                RdMeterReadsLp firstDayEnergy = energyList.get(i - 1);
                RdMeterReadsLp nextDayEnergy = energyList.get(i);
                if (nextDayEnergy.getTv() - firstDayEnergy.getTv() == 86400) {
                    double value = Convert.toDouble(nextDayEnergy.getValue()) - Convert.toDouble(firstDayEnergy.getValue());
                    value *= scale;

                    RdEnergyDaily rdEnergyDaily = new RdEnergyDaily()
                            .setMeterId(meterId)
                            .setTv(firstDayEnergy.getTv())
                            .setTvUpdate(Instant.now().getEpochSecond())
                            .setActiveImportVal(value);
                    rdEnergyDailyList.add(rdEnergyDaily);
                }
            }
        });

        if (CollectionUtils.isNotEmpty(rdEnergyDailyList)) {
            save(rdEnergyDailyList);
        }
    }

    @Override
    public boolean save(List<RdEnergyDaily> rdEnergyDailyList) {
        if(dbType.equals("oracle")) {
            return saveForOracle(rdEnergyDailyList);
        } else if(dbType.equals("postgresql")) {
            return saveForPgsql(rdEnergyDailyList);
        } else {
            log.error("unknown dbType");
            return false;
        }
    }

    private boolean saveForOracle(List<RdEnergyDaily> rdEnergyDailyList) {
        if (CollectionUtils.isEmpty(rdEnergyDailyList)) {
            return false;
        }
        log.debug("RdEnergyDaily list size {}", rdEnergyDailyList.size());

        StringBuilder sql = new StringBuilder();
        sql.append("MERGE INTO RD_ENERGY_DAILY A")
                .append(" USING (SELECT ? AS METER_ID, ? AS TV, ? AS ACTIVE_IMPORT_VAL, ? AS TV_UPDATE FROM DUAL) B")
                .append(" ON (A.METER_ID = B.METER_ID AND A.TV = B.TV)")
                .append(" WHEN MATCHED THEN UPDATE SET A.ACTIVE_IMPORT_VAL = B.ACTIVE_IMPORT_VAL, A.TV_UPDATE = B.TV_UPDATE")
                .append(" WHEN NOT MATCHED THEN INSERT(METER_ID, TV, ACTIVE_IMPORT_VAL, TV_UPDATE)")
                .append(" VALUES(B.METER_ID, B.TV, B.ACTIVE_IMPORT_VAL, B.TV_UPDATE)");

        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dnNames[0]);

        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql.toString())) {
            conn.setAutoCommit(false);
            for (RdEnergyDaily rdEnergyDaily : rdEnergyDailyList) {
                ps.setLong(1, rdEnergyDaily.getMeterId());
                ps.setLong(2, rdEnergyDaily.getTv());
                ps.setDouble(3, rdEnergyDaily.getActiveImportVal());
                ps.setLong(4, rdEnergyDaily.getTvUpdate());
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
        } catch (SQLException e) {
            log.error("batch save error", e);
            return false;
        }
        return true;
    }

    private boolean saveForPgsql(List<RdEnergyDaily> rdEnergyDailyList) {
        if (CollectionUtils.isEmpty(rdEnergyDailyList)) {
            return false;
        }
        log.debug("RdEnergyDaily list size {}", rdEnergyDailyList.size());

        StringBuilder sql = new StringBuilder();

        sql.append("with upsert as ")
                .append(" (    update RD_ENERGY_DAILY A ")
                .append(" SET ACTIVE_IMPORT_VAL = B.ACTIVE_IMPORT_VAL, TV_UPDATE = B.TV_UPDATE ")
                .append("from (SELECT ? AS METER_ID, ? AS TV, ? AS ACTIVE_IMPORT_VAL, ? AS TV_UPDATE) B ")
                .append(" where (A.METER_ID = B.METER_ID AND A.TV = B.TV) ")
                .append("returning A.METER_ID,A.TV ) ")
                .append(" insert into RD_ENERGY_DAILY(METER_ID, TV, ACTIVE_IMPORT_VAL, TV_UPDATE) ")
                .append("select * ")
                .append("from (SELECT ? AS METER_ID, ? AS TV, ? AS ACTIVE_IMPORT_VAL, ? AS TV_UPDATE) t ")
                .append("where not exists ( ")
                .append("select 1 from upsert B where t.METER_ID = B.METER_ID AND t.TV = B.TV );");
        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dnNames[0]);

        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql.toString())) {
            conn.setAutoCommit(false);
            for (RdEnergyDaily rdEnergyDaily : rdEnergyDailyList) {
                ps.setLong(1, rdEnergyDaily.getMeterId());
                ps.setLong(2, rdEnergyDaily.getTv());
                ps.setDouble(3, rdEnergyDaily.getActiveImportVal());
                ps.setLong(4, rdEnergyDaily.getTvUpdate());
                ps.setLong(5, rdEnergyDaily.getMeterId());
                ps.setLong(6, rdEnergyDaily.getTv());
                ps.setDouble(7, rdEnergyDaily.getActiveImportVal());
                ps.setLong(8, rdEnergyDaily.getTvUpdate());
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
        } catch (SQLException e) {
            log.error("batch save error", e);
            return false;
        }
        return true;
    }

    @Override
    public void reCalcDailyEnergyInterval(ReCalcDailyEnergyDTO reCalcDailyEnergyDTO) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.set("messageId", reCalcDailyEnergyDTO.getMessageId());
        try {
            List<String> meterIdstr = lineLossMapper.getMeterIdByIds(reCalcDailyEnergyDTO.getAoIds());
            if (CollectionUtils.isNotEmpty(meterIdstr)) {
                List<String> meterIdStrList = new ArrayList<>();
                List<String[]> meterArray = meterIdstr.stream()
                        .map(e -> e.split(","))
                        .collect(Collectors.toList());
                meterArray.forEach(e -> {
                    List<String> item = Arrays.asList(e);
                    if (CollectionUtils.isNotEmpty(item)) {
                        meterIdStrList.addAll(item);
                    }
                });
                List<Long> meterIds = meterIdStrList.stream()
                        .filter(e -> StrUtil.isNotEmpty(e))
                        .distinct()
                        .map(Long::valueOf)
                        .collect(Collectors.toList());
                List<RdMeterReadsLp> lastDayEnergyList = dataProcessService.getRdMeterReadLpList(meterIds, ENERGY_DATA_ITEM_ID, reCalcDailyEnergyDTO.getStartTime(), reCalcDailyEnergyDTO.getEndTime());
                this.calcDailyEnergyInterval(lastDayEnergyList, false);
                log.info("reCalcDailyEnergy size {}", lastDayEnergyList.size());
            }
            jsonObject.set("status", LogRequest.Status.success.getCode());
        } catch (Exception e) {
            e.printStackTrace();
            jsonObject.set("status", LogRequest.Status.fail.getCode());
        }
        //发送更新日志状态的kafka消息
        kafkaTemplate.send(TaskKafkaTopic.LOG_REQUEST_STATUS_UPDATE, JSONUtil.toJsonStr(jsonObject).getBytes(StandardCharsets.UTF_8));
    }

    private static <T> Collector<T, ?, List<T>> toSortedList(Comparator<? super T> c) {
        return Collectors.collectingAndThen(Collectors.toCollection(() -> new TreeSet<>(c)), ArrayList::new);
    }

    private double parseCtOrVtValue(String value) {
        double result = 1;

        if (StringUtils.isNotBlank(value)) {
            String[] parts = value.split("/");
            if (parts.length > 1) {
                double numerator = Optional.ofNullable(Convert.toDouble(parts[0])).orElse(1.0);
                double denominator = Optional.ofNullable(Convert.toDouble(parts[1])).orElse(1.0);
                result = numerator / denominator;
            } else {
                result = Optional.ofNullable(Convert.toDouble(value)).orElse(1.0);
            }
        }

        return result;
    }

}
